package com.atguigu.gmall2020.realtime.spark.app

import java.lang
import java.text.SimpleDateFormat
import java.util.Date

import com.alibaba.fastjson.{JSON, JSONObject}
import com.atguigu.gmall2020.realtime.spark.util.{KafkaUtil, OffsetManager, RedisUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.TopicPartition
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

/**
 * @deprecated:
 * @author: tiancy
 * @create: 2021/10/27
 */
/*
    TODO 编写SparkStreaming程序,通过kafka中用户行为日志中的数据,统计dau(用户日活)
      需要注意的点 : 1、判断用户今日首次登录标准 : 用户行为日志中存在page的key并且 last_page_id为null. 2、相同用户重复访问后,用户需要去重.
 */
//手动提交偏移量并从Redis中获取最后一次提交的偏移量的位置来保证精准一次性消费.
object DauAppByMapPatitionsAndObtainManually {
  def main(args: Array[String]): Unit = {


    //1.创建spark配置文件对象,指定核心数,这里指定的core需要考虑和Kafka中的分区数适配,尽可能的做到并行处理.
    val sparkConf = new SparkConf().setMaster("local[4]").setAppName("DauApp")
    //2.初始化SparkStreamingContext,指定配置对象并指定多久传输周期
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    //3.通过我们自己定义的工具类,通过Kafka数据源创建一个离散化流,处理我们存入到Kafka的主题中的用户行为日志的数据.
    val topicName = "ODS_BASE_LOG"
    val groupId = "dau_group"
    //
    /*
        TODO 获取偏移量管理器,用来读取最后一次提交的偏移量,作为初始值偏移量,这个偏移量是一个map维护[0号分区,0号分区初始值偏移量.1号分区,1号分区中的偏移量....]
          如果我们的程序从一个干净的主题中开始一直运行,没有当宕机过,这个偏移量fromOffset,从Redis中就查不到offset偏移量.就是null.走的就是else中的不需要offset的这个创建离散化流的方式创建KafkaDS.
          如果我么程序宕机了,再次启动后,从Redis中就能查到最后一次提交过的offset偏移量.这个fromOffset就不为null,会走指定偏移量位置的创建DS的方式.
     */
    val fromOffset: Map[TopicPartition, Long] = OffsetManager.getOffset(topicName, groupId)
    var KafkaDS: InputDStream[ConsumerRecord[String, String]] = null
    // TODO 如果说我们程序第一次使用手动提交偏移量的方式,Redis中是没有这个偏移量的.从Kafka中默认的offset开始
    if (fromOffset != null && fromOffset.size > 0) {
      // 通过这种手动填入偏移量的方式,获取程序最后一次挂了的时候保存的每个分区中的偏移量位置,获取到离散化流.
      KafkaDS = KafkaUtil.getKafkaStream(topicName, ssc, fromOffset, groupId)
    } else { // 如果是第一次接入Kafka,默认偏移量是没有的,需要从Kafka中的默认offset读取,从而创建离散化流.
      KafkaDS = KafkaUtil.getKafkaStream(topicName, ssc, groupId)
    }
    /*
      TODO 明确每个分区,从初始的偏移量,读了多少数据进入了当前的离散化流中.我们的偏移量到哪了.我们就可以直接在KafkaDS这个离散化流中转化成为RDD后,通过HasOffsetRanges对象获取得到.
         transform使用场景 : 如果DStream不能实现某些功能时，可以将离散化流转换为RDD来使用.这里需要转化的原因,我不能从这个DStream转化成一个对象,再拿这个对象内的具体的值,因此转化一下.
         使用Transform时,会对同一个时间段,也就是同一时间周期的微批次的RDD,都进行相同的转化操作.
     */

    var finalOfTheCurrentBatchOffset: Array[OffsetRange] = null // 当前批次的最终要提交的offset值.
    // 外边的代码,执行位置 : Driver端,并且执行一次.主要做的功能 : 直接从当前的离散化流中获取到最终的偏移量.
    val kafkaDStreamDB: DStream[ConsumerRecord[String, String]] = KafkaDS.transform {
      rdd => {
        /*
          TODO 这里的参数,代表当前微量批次中所有的RDD.执行次数 : 周期性执行,比如我们这里的周期是5秒中,也就是每个周期内创建的DStream,都会获取一次最终读取到的具体的偏移量.
            下面这种写法,会存在问题,因为我即使在这个转化内获取得到了这个 Array[OffsetRange] 所有分区中的结束点偏移量.我也出不去.不能放到整个全局中使用.因为这个最后要用这个提交偏移量的.因此要声明在外边.
            val finalOfTheCurrentBatchOffset: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 这个对象里面会存在每个分区中偏移量信息.
         */
        finalOfTheCurrentBatchOffset = rdd.asInstanceOf[HasOffsetRanges].offsetRanges // 这个对象里面会存在每个分区中偏移量信息.
        rdd
      }
    }

    /*
        TODO 开始处理我们的业务 :
          1.1、读取Kafka中的数据.将当前批次读取到的json串转化为fastjson中的通用对象.
          1.2、在获取通用对象时,还需要取出每条数据的ts并做格式转化 yyyy-MM-dd HH,最终将ts以及小时 hour再加到通用对象中.
          2.1、对我们获取的离散化流中的数据进行处理,筛选用户今日首次登录的用户行为日志信息.
            - 判断当前的json串中是否存在`page`这个属性.
              - 存在此属性,说明这条数据是页面信息,再获取当前行数据的`last_page_id`.如果last_page_id 不为null,则说明不是今天的第一次登录.舍弃当前行数据.
     */

    // 4、通过FastJson对读取的数据做通用对象处理
    val valueJsonObj: DStream[JSONObject] = kafkaDStreamDB.map {
      recode => {
        val kafkaJsonString: String = recode.value() // 获取到Kafka中存储用户行为日志数据.
        val stringTOJsonObject: JSONObject = JSON.parseObject(kafkaJsonString) // 转化为fastjson的通用对象
        val ts: lang.Long = stringTOJsonObject.getLong("ts") // 获取到通用对象的ts时间戳
        val simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH")
        val dateString: String = simpleDateFormat.format(new Date(ts))
        val dateArray: Array[String] = dateString.split(" ")
        val dt = dateArray(0) // ts = yyyy-MM-dd
        val hr = dateArray(1) //hr = HH
        stringTOJsonObject.put("dt", dt)
        stringTOJsonObject.put("hr", hr)
        stringTOJsonObject
      }
    }

    /*
        TODO 5、过滤并查找,先判断当前valueJsonObj对象中是否存在`page`的key,如果key存在,再获取last_page_id.过滤筛选今天首次登录的用户信息
           5.1、使用page属性是否存在作为过滤条件,返回存在page的通用对象 existPageValueJsonObj
           5.2、判断当前page对象中是否存在last_page_id 是否为null,如果为 null则是今日首次.否则过滤掉
           5.3、将今日首日登录的用户对应的设备id存入到redis中.来去重.
     */
    val existPageValueJsonObj: DStream[JSONObject] = valueJsonObj.filter(_.containsKey("page")) // 过滤掉不存在page属性的json通用对象
    val firstLoginUser: DStream[JSONObject] = existPageValueJsonObj.filter { // 过滤并获取今日首次登录app的用户信息.
      jsonObj => {
        var flag = false
        val lastPageId: String = jsonObj.getJSONObject("page").getString("last_page_id")
        if (lastPageId != null && lastPageId.nonEmpty) {
          flag
        } else {
          !flag // 此条数据包含了page属性且存在 last_page_id == null(存在访问日志,并且没有上一页)
        }
      }
    }
    //firstLoginUser.print(1000)

    /*
        TODO 5.3、将今日首日登录的用户对应的设备id存入到redis中.去重.首日登录的用户具体的信息,依然存在离散化流中.等待写入到 olap这种容器中等待分析(这里用es来做)
          - 这里依然需要考虑使用什么算子,我们往redis中存的时候,一定会出现重复性问题,如果已经存在redis中,则当前行数据过滤掉.如果没有存,则当前行数据中设备id直接存到redis,当前行数据留在离散化流中
            基于上述的分析,我们这里使用的算子,一定要有过滤功能,(进入的行数和执行结束的行数不一致) filter 或者 mapPartitions
          - 在redis中如何存 ? 用什么来存呢 ? 回顾redis中的数据结构 : String、List、Set、Hash、zset(存入数据不需要排序,排除掉zset)(需要按照年-月-日+标志作为key,每天的存入到一个列表中,还剩下一个list一个set) ==> 需要去除重复,使用set
     */

    val activeUserInfoDStream: DStream[JSONObject] = firstLoginUser.mapPartitions {
      jsonObjItr => {
        //每分区 每批次的数据,这里传入的参数是一个迭代器.也要返回一个迭代器.
        val jedis: Jedis = RedisUtil.getJedisClient //分区共用
        val originList: List[JSONObject] = jsonObjItr.toList //算子中每个分区内的原始数据
        val activeUserInfoList = new ListBuffer[JSONObject] // 创建一个可变集合,用来存放当日活跃用户的信息
        println("该分区共：" + originList.size + "条")
        for (jsonObj <- originList) {
          val mid: String = jsonObj.getJSONObject("common").getString("mid")
          // 构建set中的key ==> dau:2021-10-29 (dau:yyyy-MM-dd),value ==> mid
          val dt: String = jsonObj.getString("dt")
          val key: String = "dau:" + dt
          val depositMarker: lang.Long = jedis.sadd(key, mid) //是否存入标记. 1存入成功(存入成功,说明用户首次登录,数据保留) . 0 存入失败(用户今日不是首日登录,数据过滤掉)
          if (depositMarker == 1L) {
            //如果写入redis成功(1L),则流中的数据保存.否则不保存.流中的数据,应该存到一个容器中.这里使用可变的List ==> activeUserInfoList
            activeUserInfoList.append(jsonObj)
          }
        }
        jedis.close()
        println("该分区筛选后共：" + activeUserInfoList.size + "条")
        activeUserInfoList.iterator
      }
    }

    //activeUserInfoDStream.print(1000)
    /*
      TODO 向olap中写入数据并从driver中提交偏移量.
        foreachRDD和transform区别 : 看最终返回的结果,是否还需要处理,如果不需要处理,则执行使用foreachRDD,如果还需要返回DStream进行下一步操作,则使用transform转化操作.
     */
    activeUserInfoDStream.foreachRDD {
      rdd => {
        //周期性执行的代码
        rdd.foreachPartition {
          //算子内的代码,需要在executor端执行
          dauIter =>
            println(dauIter.toList.mkString("\n"))
        }
        /*
          TODO 开始手动提交当前批次的最终偏移量 finalOfTheCurrentBatchOffset,这里需要调用具体的写入Redis中的方法.
            提交偏移量的时机 : 每个周期内通过driver端执行.
         */
        OffsetManager.saveOffset(topicName, groupId, finalOfTheCurrentBatchOffset)
      }
    }
    ssc.start()
    ssc.awaitTermination()
  }
}
